package com.atguigu.gmall.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionCommentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionCommentInfo().start(10012, 4, Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // TODO 从kafka的topic_db主题获取数据，并创建动态表---kafka连接器
        readOdsDb(tableEnv, Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
//        tableEnv.executeSql("select * from topic_db").print();

        // TODO 过滤数据 type = 'insert' & table = 'comment_info'
        Table commentInfo = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['appraise'] appraise,\n" +
                "    `data`['comment_txt'] comment_txt,\n" +
                "    ts,\n" +
                "    pt\n" +
                "from topic_db\n" +
                "where `table` = 'comment_info' and `type` = 'insert'");
        tableEnv.createTemporaryView("comment_info",commentInfo);
        //tableEnv.executeSql("select * from comment_info").print();

        // TODO 从Hbase获取dic信息，并创建动态表---Hbase连接器
        readBaseDic(tableEnv);
        //tableEnv.executeSql("select dic_code,dic_name from base_dic").print();

        // TODO 关联字典表和评论数据--lookup join
        Table joinedTable = tableEnv.sqlQuery("SELECT \n" +
                "   id,user_id,sku_id,appraise,dic.dic_name as appraise_name,comment_txt,ts\n" +
                "   FROM comment_info AS c\n" +
                "   JOIN base_dic FOR SYSTEM_TIME AS OF c.pt AS dic\n" +
                "   ON c.appraise = dic.dic_code");
        //joinedTable.execute().print();

        // TODO 将关联数据写入kafka主题中---upsert-kafka连接器
        tableEnv.executeSql("CREATE TABLE " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO +" (\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    sku_id string,\n" +
                "    appraise string,\n" +
                "    appraise_name string,\n" +
                "    comment_txt string,\n" +
                "    ts bigint,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
        joinedTable.executeInsert(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
    }
}
